package com.tpvlog.dfs.namenode.server;

import com.tpvlog.dfs.namenode.file.FSNameSystem;
import com.tpvlog.dfs.namenode.register.DataNodeManager;
import com.tpvlog.dfs.rpc.service.*;
import io.grpc.stub.StreamObserver;

/**
 * NameNode的RPC服务接口
 *
 * @author Ressmix
 */
public class NameNodeServiceImpl extends NameNodeServiceGrpc.NameNodeServiceImplBase {

    public static final Integer STATUS_SUCCESS = 1;
    public static final Integer STATUS_FAILURE = 2;

    /**
     * 负责管理元数据的核心组件
     */
    private FSNameSystem namesystem;

    /**
     * 负责管理集群中所有的datanode的组件
     */
    private DataNodeManager datanodeManager;

    public NameNodeServiceImpl(FSNameSystem namesystem, DataNodeManager datanodeManager) {
        this.namesystem = namesystem;
        this.datanodeManager = datanodeManager;
    }

    /**
     * DataNode注册
     */
    public void register(RegisterRequest request, StreamObserver<RegisterResponse> responseObserver) {
        // 使用DataNodeManager组件完成DataNode注册
        datanodeManager.register(request.getIp(), request.getHostname());

        RegisterResponse response = RegisterResponse.newBuilder().setStatus(STATUS_SUCCESS).build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * DataNode心跳
     */
    public void heartbeat(HeartbeatRequest request, StreamObserver<HeartbeatResponse> responseObserver) {
        // 使用DataNodeManager组件完成DataNode心跳
        datanodeManager.heartbeat(request.getIp(), request.getHostname());

        HeartbeatResponse response = HeartbeatResponse.newBuilder().setStatus(STATUS_SUCCESS).build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 创建目录
     */
    @Override
    public void mkdir(MkDirRequest request, StreamObserver<MkDirResponse> responseObserver) {
        try {
            this.namesystem.mkdir(request.getPath());

            System.out.println("创建目录：path" + request.getPath());

            MkDirResponse response = MkDirResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
